/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.tool;

import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.sss.client.DataGetter;
import org.sss.client.SssClient;
import org.sss.mapreduce.Cluster;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.util.BytesComparator;
import org.sss.util.Hexdump;

public class TupleGroupCompare {
  private boolean invert;
  private boolean debug;
  private GroupID gid0;
  private GroupID gid1;
  private List<StorageNode> storageNodes;
  private final PrintStream ps;
  private final SssClient client;

  private TupleGroupCompare(SssClient client) {
    this.invert = false;
    this.debug  = false;
    this.storageNodes = null;
    this.ps = new PrintStream(System.out);
    this.client = client;
  }

  private void usage(final Options options) {
    HelpFormatter formatter = new HelpFormatter();
    formatter.printHelp("TupleGroupCompare [options] gid0 gid1", options);
    System.exit(-1);
  }

  private void parseArgs(String[] args) {
    Options options = new Options();
    options.addOption("d", false, "debug option");
    options.addOption("v", false, "invert match");
    OptionBuilder.withArgName("nodelist");
    OptionBuilder.hasArg();
    OptionBuilder.withDescription("use specified nodes only " +
                                  "(node index separated by comma)");
    Option node = OptionBuilder.create("n");
    options.addOption(node);
    CommandLineParser parser = new BasicParser();
    CommandLine commandLine = null;
    int[] nodeIndex = null;
    try {
      commandLine = parser.parse(options, args);
      if (commandLine.hasOption("d")) { debug  = true; }
      if (commandLine.hasOption("v")) { invert = true; }
      if (commandLine.hasOption("n")) {
        String[] nodeList = commandLine.getOptionValues("n")[0].split(",");
        nodeIndex = new int[nodeList.length];
        int i = 0;
        for (String n: nodeList) {
          nodeIndex[i++] = Integer.parseInt(n);
        }
      }
      String[] gidList = commandLine.getArgs();
      if (gidList.length != 2) {
        System.err.println(gidList.length < 2 ?
                           "no GroupID specified." :
                           "too many GrpupIDs specified.");
        usage(options);
      }
      gid0 = GroupID.createFromString(gidList[0]);
      gid1 = GroupID.createFromString(gidList[1]);
    }
    catch (ParseException e) {
      usage(options);
    }
    catch (IllegalArgumentException e) {
      System.err.println("invalid GroupID string specified.");
      usage(options);
    }

    try {
      Cluster cluster;
      cluster = Cluster.loadDefaultConfigFile();
      this.storageNodes = new ArrayList<StorageNode>();
      if (nodeIndex != null) {
        for (int i=0; i<nodeIndex.length; i++) {
          this.storageNodes.add(cluster.getStorageNodes().get(nodeIndex[i]));
        }
      }
      else {
        this.storageNodes.addAll(cluster.getStorageNodes());
      }
    }
    catch (IndexOutOfBoundsException e) {
      System.err.println("specified node index is out of range.");
      System.exit(-1);
    }
    catch (Exception e) {
      System.err.println("cannot load cluster configuration file.");
      System.exit(-1);
    }
  }
  private <T> Iterator<T> sortedIterator(Iterator<T> it, Comparator<T> comparator) {
    List<T> list = new ArrayList<T>();
    while (it.hasNext()) {
      list.add(it.next());
    }
    Collections.sort(list, comparator);
    return list.iterator();
  }
  private Comparator<Tuple<PackableBytes, PackableBytes>> keyValueComparator() {
    return new Comparator<Tuple<PackableBytes, PackableBytes>>() {
      private final BytesComparator cmp = new BytesComparator();
      @Override
      public int compare(Tuple<PackableBytes, PackableBytes> o1,
                         Tuple<PackableBytes, PackableBytes> o2) {
        int keyCmp = cmp.compare(o1.getKey().get(), o2.getKey().get());
        if (keyCmp == 0) {
          return cmp.compare(o1.getValue().get(), o2.getValue().get());
        }
        else {
          return keyCmp;
        }
      }
      @Override
      public boolean equals(Object o) {
        return (this == o);
      }
    };
  }
  private void printKeyValue(Tuple<PackableBytes, PackableBytes> kv) {
    System.out.println("hexdump of raw key data:");
    Hexdump.hexdump(ps, kv.getKey().get(), 0, kv.getKey().get().length);
    System.out.println("hexdump of raw value data:");
    Hexdump.hexdump(ps, kv.getValue().get(), 0, kv.getValue().get().length);
  }
  private void start() throws Exception {
    if (debug) {
      System.out.println("Compare following GroupIDs:");
      System.out.println("  gid0: " + gid0.toString());
      System.out.println("  gid1: " + gid1.toString());
      System.out.println("Target storage node(s): " + storageNodes.size());
    }
    DataGetter<PackableBytes, PackableBytes>
      dg0 = DataGetter.create(client, PackableBytes.class, PackableBytes.class,
                              gid0.getNonVolatileCopy(), storageNodes);
    DataGetter<PackableBytes, PackableBytes>
      dg1 = DataGetter.create(client, PackableBytes.class, PackableBytes.class,
                              gid1.getNonVolatileCopy(), storageNodes);
    Iterator<Tuple<PackableBytes, PackableBytes>>
      it0 = sortedIterator(dg0.iterator(), keyValueComparator());
    Iterator<Tuple<PackableBytes, PackableBytes>>
      it1 = sortedIterator(dg1.iterator(), keyValueComparator());
    final int EXIT  = -1;
    final int LEFT  =  0;
    final int RIGHT =  1;
    final int BOTH  =  2;
    int mode = BOTH;
    Tuple<PackableBytes, PackableBytes> kv0 = null;
    Tuple<PackableBytes, PackableBytes> kv1 = null;
    while (mode != EXIT) {
      if (mode == BOTH || mode == LEFT) {
        if (it0.hasNext()) {
          kv0 = it0.next();
        }
        else {
          kv0 =  null;
        }
      }
      if (mode == BOTH || mode == RIGHT) {
        if (it1.hasNext()) {
          kv1 = it1.next();
        }
        else {
          kv1 = null;
        }
      }
      if (kv0 != null && kv1 != null) {
        int cmpkv = keyValueComparator().compare(kv0, kv1);
        if (cmpkv < 0) {
          if (!invert) {
            System.out.println("Only exists in " + gid0.toString());
            printKeyValue(kv0);
          }
          mode = LEFT;
        }
        else if (cmpkv == 0) {
          if (invert || debug) {
            System.out.println("Found match");
            printKeyValue(kv0);
          }
          mode = BOTH;
        }
        else {                  // cmpkv > 0
          if (!invert) {
            System.out.println("Only exists in " + gid1.toString());
            printKeyValue(kv1);
          }
          mode = RIGHT;
        }
      }
      else if (kv0 != null && kv1 == null) {
        if (!invert) {
          System.out.println("Only exists in " + gid0.toString());
          printKeyValue(kv0);
          while (it0.hasNext()) {
            printKeyValue(it0.next());
          }
        }
        mode = EXIT;
      }
      else if (kv0 == null && kv1 != null) {
        if (!invert) {
          System.out.println("Only exists in " + gid1.toString());
          printKeyValue(kv1);
          while (it1.hasNext()) {
            printKeyValue(it1.next());
          }
        }
        mode = EXIT;
      }
      else {
        mode = EXIT;
      }
    }
    dg0.close(); // don't forget this.
    dg1.close(); // don't forget this.
  }
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    TupleGroupCompare app = new TupleGroupCompare(client);
    app.parseArgs(client.getArgs());
    app.start();
    System.exit(0);
  }
}
